package sync

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/louloulin/dataflare/pkg/state/store"
)

// SyncMode represents the synchronization mode
type SyncMode int

const (
	// SyncModeAsync represents asynchronous synchronization
	SyncModeAsync SyncMode = iota
	// SyncModeSync represents synchronous synchronization
	SyncModeSync
	// SyncModeSemiSync represents semi-synchronous synchronization
	SyncModeSemiSync
)

// String returns the string representation of the sync mode
func (m SyncMode) String() string {
	switch m {
	case SyncModeAsync:
		return "async"
	case SyncModeSync:
		return "sync"
	case SyncModeSemiSync:
		return "semi-sync"
	default:
		return "unknown"
	}
}

// SyncStrategy represents the synchronization strategy
type SyncStrategy int

const (
	// SyncStrategyFull represents full synchronization
	SyncStrategyFull SyncStrategy = iota
	// SyncStrategyIncremental represents incremental synchronization
	SyncStrategyIncremental
	// SyncStrategyDelta represents delta synchronization
	SyncStrategyDelta
)

// String returns the string representation of the sync strategy
func (s SyncStrategy) String() string {
	switch s {
	case SyncStrategyFull:
		return "full"
	case SyncStrategyIncremental:
		return "incremental"
	case SyncStrategyDelta:
		return "delta"
	default:
		return "unknown"
	}
}

// SyncConfig represents the synchronization configuration
type SyncConfig struct {
	// Mode is the synchronization mode
	Mode SyncMode
	// Strategy is the synchronization strategy
	Strategy SyncStrategy
	// Interval is the synchronization interval
	Interval time.Duration
	// Timeout is the synchronization timeout
	Timeout time.Duration
	// RetryCount is the retry count
	RetryCount int
	// RetryInterval is the retry interval
	RetryInterval time.Duration
	// BatchSize is the batch size
	BatchSize int
	// Nodes is the list of nodes to synchronize with
	Nodes []string
}

// DefaultSyncConfig returns the default synchronization configuration
func DefaultSyncConfig() *SyncConfig {
	return &SyncConfig{
		Mode:          SyncModeAsync,
		Strategy:      SyncStrategyIncremental,
		Interval:      time.Second * 5,
		Timeout:       time.Second * 10,
		RetryCount:    3,
		RetryInterval: time.Second,
		BatchSize:     100,
		Nodes:         []string{},
	}
}

// SyncOperation represents a synchronization operation
type SyncOperation struct {
	// Type is the operation type
	Type string
	// Key is the key
	Key string
	// Value is the value
	Value []byte
	// Timestamp is the timestamp
	Timestamp time.Time
	// Source is the source node
	Source string
}

// SyncTransport represents the synchronization transport interface
type SyncTransport interface {
	// Connect connects to a node
	Connect(ctx context.Context, node string) error
	// Disconnect disconnects from a node
	Disconnect(ctx context.Context, node string) error
	// SendOperations sends operations to a node
	SendOperations(ctx context.Context, node string, operations []*SyncOperation) error
	// ReceiveOperations receives operations from a node
	ReceiveOperations(ctx context.Context) ([]*SyncOperation, string, error)
	// SendAck sends an acknowledgment to a node
	SendAck(ctx context.Context, node string, timestamp time.Time) error
	// ReceiveAck receives an acknowledgment from a node
	ReceiveAck(ctx context.Context) (string, time.Time, error)
}

// SyncManager represents the synchronization manager
type SyncManager struct {
	// config is the configuration
	config *SyncConfig
	// transport is the transport
	transport SyncTransport
	// store is the state store
	store store.StateStore
	// operations is the operation queue
	operations []*SyncOperation
	// lastSync is the last synchronization time
	lastSync time.Time
	// mu is the mutex
	mu sync.RWMutex
	// ctx is the context
	ctx context.Context
	// cancel is the cancel function
	cancel context.CancelFunc
	// wg is the wait group
	wg sync.WaitGroup
}

// NewSyncManager creates a new synchronization manager
func NewSyncManager(config *SyncConfig, transport SyncTransport, store store.StateStore) *SyncManager {
	if config == nil {
		config = DefaultSyncConfig()
	}

	ctx, cancel := context.WithCancel(context.Background())

	return &SyncManager{
		config:     config,
		transport:  transport,
		store:      store,
		operations: make([]*SyncOperation, 0),
		lastSync:   time.Now(),
		ctx:        ctx,
		cancel:     cancel,
	}
}

// Start starts the synchronization manager
func (m *SyncManager) Start() error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Connect to all nodes
	for _, node := range m.config.Nodes {
		err := m.transport.Connect(m.ctx, node)
		if err != nil {
			return fmt.Errorf("failed to connect to node %s: %w", node, err)
		}
	}

	// Start the synchronization loop
	m.wg.Add(1)
	go m.syncLoop()

	// Start the receive loop
	m.wg.Add(1)
	go m.receiveLoop()

	return nil
}

// Stop stops the synchronization manager
func (m *SyncManager) Stop() error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Cancel the context
	m.cancel()

	// Wait for all goroutines to finish
	m.wg.Wait()

	// Disconnect from all nodes
	for _, node := range m.config.Nodes {
		err := m.transport.Disconnect(m.ctx, node)
		if err != nil {
			return fmt.Errorf("failed to disconnect from node %s: %w", node, err)
		}
	}

	return nil
}

// AddOperation adds an operation to the queue
func (m *SyncManager) AddOperation(op *SyncOperation) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.operations = append(m.operations, op)
}

// GetLastSyncTime returns the last synchronization time
func (m *SyncManager) GetLastSyncTime() time.Time {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.lastSync
}

// syncLoop is the synchronization loop
func (m *SyncManager) syncLoop() {
	defer m.wg.Done()

	ticker := time.NewTicker(m.config.Interval)
	defer ticker.Stop()

	for {
		select {
		case <-m.ctx.Done():
			return
		case <-ticker.C:
			m.sync()
		}
	}
}

// sync performs synchronization
func (m *SyncManager) sync() {
	m.mu.Lock()
	// Check if there are operations to synchronize
	if len(m.operations) == 0 {
		m.mu.Unlock()
		return
	}

	// Get operations to synchronize
	batchSize := m.config.BatchSize
	if batchSize > len(m.operations) {
		batchSize = len(m.operations)
	}
	operations := make([]*SyncOperation, batchSize)
	copy(operations, m.operations[:batchSize])
	m.mu.Unlock()

	// Synchronize with all nodes
	for _, node := range m.config.Nodes {
		m.syncWithNode(node, operations)
	}

	// Update the operation queue
	m.mu.Lock()
	m.operations = m.operations[batchSize:]
	m.lastSync = time.Now()
	m.mu.Unlock()
}

// syncWithNode synchronizes with a node
func (m *SyncManager) syncWithNode(node string, operations []*SyncOperation) {
	// Create a context with timeout
	ctx, cancel := context.WithTimeout(m.ctx, m.config.Timeout)
	defer cancel()

	// Send operations
	err := m.transport.SendOperations(ctx, node, operations)
	if err != nil {
		fmt.Printf("Failed to send operations to node %s: %v\n", node, err)
		return
	}

	// If synchronous mode, wait for acknowledgment
	if m.config.Mode == SyncModeSync {
		for i := 0; i < m.config.RetryCount; i++ {
			// Receive acknowledgment
			ackNode, timestamp, err := m.transport.ReceiveAck(ctx)
			if err != nil {
				// Retry
				time.Sleep(m.config.RetryInterval)
				continue
			}

			// Check if the acknowledgment is from the correct node
			if ackNode == node {
				// Update the last synchronization time
				m.mu.Lock()
				m.lastSync = timestamp
				m.mu.Unlock()
				return
			}
		}

		fmt.Printf("Failed to receive acknowledgment from node %s\n", node)
	}
}

// receiveLoop is the receive loop
func (m *SyncManager) receiveLoop() {
	defer m.wg.Done()

	for {
		select {
		case <-m.ctx.Done():
			return
		default:
			// Receive operations
			operations, node, err := m.transport.ReceiveOperations(m.ctx)
			if err != nil {
				// Check if the context is canceled
				if m.ctx.Err() != nil {
					return
				}
				// Log the error and continue
				fmt.Printf("Failed to receive operations: %v\n", err)
				time.Sleep(m.config.RetryInterval)
				continue
			}

			// Process operations
			m.processOperations(operations)

			// Send acknowledgment
			if m.config.Mode == SyncModeSync {
				err = m.transport.SendAck(m.ctx, node, time.Now())
				if err != nil {
					fmt.Printf("Failed to send acknowledgment to node %s: %v\n", node, err)
				}
			}
		}
	}
}

// processOperations processes operations
func (m *SyncManager) processOperations(operations []*SyncOperation) {
	for _, op := range operations {
		switch op.Type {
		case "set":
			// Get the state
			state, err := m.store.GetState(m.ctx, "default")
			if err != nil {
				fmt.Printf("Failed to get state: %v\n", err)
				continue
			}

			// Check the state type
			kvState, ok := state.(store.KVState)
			if !ok {
				fmt.Printf("State is not a KV state\n")
				continue
			}

			// Set the value
			err = kvState.Set(m.ctx, op.Key, op.Value)
			if err != nil {
				fmt.Printf("Failed to set value: %v\n", err)
				continue
			}
		case "delete":
			// Get the state
			state, err := m.store.GetState(m.ctx, "default")
			if err != nil {
				fmt.Printf("Failed to get state: %v\n", err)
				continue
			}

			// Check the state type
			kvState, ok := state.(store.KVState)
			if !ok {
				fmt.Printf("State is not a KV state\n")
				continue
			}

			// Delete the value
			err = kvState.Delete(m.ctx, op.Key)
			if err != nil {
				fmt.Printf("Failed to delete value: %v\n", err)
				continue
			}
		default:
			fmt.Printf("Unknown operation type: %s\n", op.Type)
		}
	}
}

// HTTPSyncTransport represents the HTTP synchronization transport
type HTTPSyncTransport struct {
	// baseURL is the base URL
	baseURL string
	// client is the HTTP client
	client *http.Client
	// operationCh is the operation channel
	operationCh chan *OperationMessage
	// ackCh is the acknowledgment channel
	ackCh chan *AckMessage
}

// OperationMessage represents an operation message
type OperationMessage struct {
	// Node is the node
	Node string
	// Operations is the operations
	Operations []*SyncOperation
}

// AckMessage represents an acknowledgment message
type AckMessage struct {
	// Node is the node
	Node string
	// Timestamp is the timestamp
	Timestamp time.Time
}

// NewHTTPSyncTransport creates a new HTTP synchronization transport
func NewHTTPSyncTransport(baseURL string) *HTTPSyncTransport {
	return &HTTPSyncTransport{
		baseURL:     baseURL,
		client:      &http.Client{},
		operationCh: make(chan *OperationMessage, 100),
		ackCh:       make(chan *AckMessage, 100),
	}
}

// Connect connects to a node
func (t *HTTPSyncTransport) Connect(ctx context.Context, node string) error {
	// In a real implementation, this would establish an HTTP connection
	return nil
}

// Disconnect disconnects from a node
func (t *HTTPSyncTransport) Disconnect(ctx context.Context, node string) error {
	// In a real implementation, this would close the HTTP connection
	return nil
}

// SendOperations sends operations to a node
func (t *HTTPSyncTransport) SendOperations(ctx context.Context, node string, operations []*SyncOperation) error {
	// In a real implementation, this would send operations via HTTP
	// For testing, we send to the operation channel
	select {
	case t.operationCh <- &OperationMessage{Node: node, Operations: operations}:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// ReceiveOperations receives operations from a node
func (t *HTTPSyncTransport) ReceiveOperations(ctx context.Context) ([]*SyncOperation, string, error) {
	// In a real implementation, this would receive operations via HTTP
	// For testing, we receive from the operation channel
	select {
	case msg := <-t.operationCh:
		return msg.Operations, msg.Node, nil
	case <-ctx.Done():
		return nil, "", ctx.Err()
	}
}

// SendAck sends an acknowledgment to a node
func (t *HTTPSyncTransport) SendAck(ctx context.Context, node string, timestamp time.Time) error {
	// In a real implementation, this would send an acknowledgment via HTTP
	// For testing, we send to the acknowledgment channel
	select {
	case t.ackCh <- &AckMessage{Node: node, Timestamp: timestamp}:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// ReceiveAck receives an acknowledgment from a node
func (t *HTTPSyncTransport) ReceiveAck(ctx context.Context) (string, time.Time, error) {
	// In a real implementation, this would receive an acknowledgment via HTTP
	// For testing, we receive from the acknowledgment channel
	select {
	case msg := <-t.ackCh:
		return msg.Node, msg.Timestamp, nil
	case <-ctx.Done():
		return "", time.Time{}, ctx.Err()
	}
}

// Import http package
import "net/http"
